package zhang;

import app.BaseSqlApp;
import common.Constant;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import util.SQLUtil;

public class DwdTestTest extends BaseSqlApp {
    public static void main(String[] args) {
        new app.dwd.DwdTestTest().init(3010,2,"DwdTestTest");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        //1 读取ods数据
        readOdsDb(tEnv,"DwdTestTest");

/*
{
  "database": "edu",
  "table": "test_exam",
  "type": "insert",
  "ts": 1652667828,
  "xid": 2851,
  "commit": true,
  "data": {
    "id": 2363,
    "paper_id": 189,
    "user_id": 1079,
    "score": 78.40,
    "duration_sec": 1157,
    "create_time": "2022-02-22 21:42:16",
    "submit_time": "2022-02-22 21:42:16",
    "update_time": null,
    "deleted": "0"
  }
}
 */


        // 2 过滤测验表
        Table testExam = tEnv.sqlQuery(
                "select " +
                        "`data`['id'] id, " +
                        "cast(`data`['paper_id'] as bigint) paper_id, " +
                        "`data`['user_id'] user_id, " +
                        "`data`['score'] score, " +
                        "`data`['duration_sec'] duration_sec, " +
                        "`data`['create_time'] create_time, " +
                        "`data`['submit_time'] submit_time, " +
                        "`data`['update_time'] update_time, " +
                        "ts, " +
                        "pt " +
                        "from ods_db " +
                        "where `database`='edu' " +
                        "and `table`='test_exam' " +
                        "and `type`='insert' ");
        tEnv.createTemporaryView("test_exam", testExam);



/*

{
  "database": "edu",
  "table": "test_paper",
  "type": "bootstrap-insert",
  "ts": 1652681617,
  "data": {
    "id": 322,
    "paper_title": "尚硅谷AJAX技术测试",
    "course_id": 379,
    "create_time": "2022-08-31 21:00:02",
    "update_time": null,
    "publisher_id": 99,
    "deleted": "0"
  }
}
 */

        // 3 退化试卷表中的课程字段
        tEnv.executeSql("create table test_paper(" +
                "id bigint, " +
                "course_id bigint " +
                ")with(" +
                " 'connector'='jdbc',  " +
                " 'url' = 'jdbc:mysql://hadoop162:3306/edu?useSSL=false'," +
                " 'table-name' = 'test_paper', " +
                " 'username' = 'root',  " +
                " 'password' = 'aaaaaa', " +
                " 'lookup.cache.max-rows' = '50', " +
                " 'lookup.cache.ttl' = '1 hour' " +
                ")");

        //4  退化维度
        Table result = tEnv.sqlQuery("select " +
                "te.id, " +
                "te.paper_id, " +
                "te.user_id, " +
                "cast(cast(te.score as decimal(10,2)) as string) score, " +
                "cast(te.duration_sec as bigint) duration_sec, " +
                "case " +
                "           when te.score >= 90 then '90-100' " +
                "           when te.score >= 80 then '80-90' " +
                "           when te.score >= 70 then '70-80' " +
                "           when te.score >= 60 then '60-70' " +
                "           when te.score >= 0 then '0-60' " +
                "    end point_rank, " +
                "date_format(te.create_time, 'yyyy-MM-dd') date_id, " +
                "te.create_time, " +
                "date_format(te.submit_time, 'yyyy-MM-dd') submit_time_date_id, " +
                "te.submit_time, " +
                "date_format(te.update_time, 'yyyy-MM-dd') update_time_date_id, " +
                "te.update_time," +
                "tp.course_id, " +
                "te.ts " +
                "from test_exam te " +
                "join test_paper for system_time as of te.pt as tp " +
                "on te.paper_id=tp.id ");


        //5 定义动态与输出的topic关联
        tEnv.executeSql("create table dwd_test_test( " +
                "id string, " +
                "paper_id bigint, " +
                "user_id string, " +
                "score string, " +
                "duration_sec bigint, " +
                "point_rank string, " +
                "date_id string, " +
                "create_time string, " +
                "submit_time_date_id string, " +
                "submit_time string, " +
                "update_time_date_id string, " +
                "update_time string, " +
                "course_id bigint, " +
                "ts bigint " +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TEST_TEST));



        // 6   把join的结果写出到输出的表
        result.executeInsert("dwd_test_test");


    }
}

